package com.dsk.connector.source.utils;

import cn.hutool.core.date.DateUtil;
import com.dsk.connector.commons.trans.FieldSchema;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;

import java.time.LocalDate;
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.*;

public class SourceRecordUtils {

    public static <T> T getStructValue(Struct struct, String fieldName){
        Schema connectSchema=(Schema)struct.get("schema");
        Field field=connectSchema.field(fieldName);
        T value=(T)struct.getArray("values").get(field.index());

        return value;
    }

    public static List<FieldSchema> convertToFieldSchema(Schema schema) {
        List<FieldSchema> fieldSchemas=new LinkedList<>();
        for(Field field: schema.fields()){
            FieldSchema fieldSchema=new FieldSchema();
            fieldSchema.setField(field.name());
            fieldSchema.setType(field.schema().type().getName());
            fieldSchema.setName(field.schema().name());
            fieldSchema.setOptional(field.schema().isOptional());
            fieldSchema.setVersion(field.schema().version());
            fieldSchema.setIndex(field.index());

            fieldSchemas.add(fieldSchema);
        }

        return fieldSchemas;
    }

    /**
     * 根据payload中的结构获取数据Map
     * @param payloadStruct
     * @return
     */
    public static Map<String, Object> convertFieldValues(Struct payloadStruct) {
        Map<String,Object> valueMap=new HashMap<>();
        for(Field field:payloadStruct.schema().fields()){
            Object value=payloadStruct.get(field);
            if(value==null || StringUtils.isBlank(value.toString())){
                valueMap.put(field.name(),value);
                continue;
            }
            if("io.debezium.time.Timestamp".equals(field.schema().name()) ){
                Date dateValue=new Date((Long)value);
                value=DateUtil.format(dateValue,"yyyy-MM-dd HH:mm:ss");
            }
            if("io.debezium.time.Date".equals(field.schema().name()) ){
                LocalDate localDate=LocalDate.ofEpochDay((Integer)value);
                value=localDate.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
            }
            if("io.debezium.time.Time".equals(field.schema().name()) ){
                LocalTime localtime=LocalTime.ofSecondOfDay((Integer)value/1000);
                value=localtime.format(DateTimeFormatter.ofPattern("HH:mm:ss"));
            }
            valueMap.put(field.name(),value);
        }
        return valueMap;
    }
}
